package org.xq.softcup.mq.client.factory;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.xq.softcup.mq.client.broker.IMqBroker;
import org.xq.softcup.mq.client.consumer.IMqConsumer;
import org.xq.softcup.mq.client.consumer.annotation.MqConsumer;
import org.xq.softcup.mq.client.consumer.ConsumerRegistrySupport;
import org.xq.softcup.mq.client.consumer.ConsumerThread;
import org.xq.softcup.mq.client.message.MqMessage;
import org.xq.softcup.mq.client.producer.MsgCallBack;
import org.xq.softcup.mq.registry.impl.XQRegistryServiceRegistry;
import org.xq.softcup.mq.remoting.invoker.XQRpcInvokerFactory;
import org.xq.softcup.mq.remoting.invoker.call.CallType;
import org.xq.softcup.mq.remoting.invoker.reference.XQRpcReferenceBean;
import org.xq.softcup.mq.remoting.invoker.router.LoadBalance;
import org.xq.softcup.mq.remoting.net.NetEnum;
import org.xq.softcup.mq.serialize.Serializer;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

/**
 * MQ客户端工厂类
 * @author admin
 */
@Slf4j
public class MqClientFactory implements ApplicationContextAware, DisposableBean {

    private String adminAddress;
    private String accessToken;
    private List<IMqConsumer> consumerList;
    private AtomicLong backMsgCount = new AtomicLong(0l);
    // 消费者线程集合
    private List<ConsumerThread> consumerThreadList = new ArrayList<ConsumerThread>();

    public void setAdminAddress(String adminAddress) {
        this.adminAddress = adminAddress;
    }
    public void setAccessToken(String accessToken) {
        this.accessToken = accessToken;
    }
    public void setConsumerList(List<IMqConsumer> consumerList) {
        this.consumerList = consumerList;
    }

    //Bean 实例化调用方法
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

        // 加载消费者
        List<IMqConsumer> consumerList = new ArrayList<>();
        //扫描@MyConsumer注解, 加入消费者列表
        Map<String, Object> serviceMap = applicationContext.getBeansWithAnnotation(MqConsumer.class);
        if (serviceMap!=null && serviceMap.size()>0) {
            for (Object serviceBean : serviceMap.values()) {
                if (serviceBean instanceof IMqConsumer) {
                    consumerList.add((IMqConsumer) serviceBean);
                }
            }
        }
        //添加消费者列表
        setConsumerList(consumerList);
        //初始化
        init();
    }

    public void init() {
        //检查
        validConsumer();
        //启动Broker服务
        startBrokerService();
        // 启动消费者
        startConsumer();
    }
    //Spring 销毁方法
    @Override
    public void destroy() throws Exception {
        destoryClientFactoryThreadPool();
        // 销毁消费者
        destoryConsumer();
        // 销毁Broker
        destoryBrokerService();

    }
    //线程池
    private ExecutorService clientFactoryThreadPool = Executors.newCachedThreadPool();
    public static volatile boolean clientFactoryPoolStoped = false;
    /**
     * 销毁消费者线程
     */
    private void destoryClientFactoryThreadPool(){
        clientFactoryPoolStoped = true;
        clientFactoryThreadPool.shutdownNow();
    }
    private XQRpcInvokerFactory xqRpcInvokerFactory = null;

    private static IMqBroker mqBroker;
    //消费者注册帮助类
    private static ConsumerRegistrySupport consumerRegistrySupport = null;
    //内存队列存放新消息
    private static LinkedBlockingQueue<MqMessage> newMessageQueue = new LinkedBlockingQueue<>();
    //回调消息存放
    private static LinkedBlockingQueue<MqMessage> callbackMessageQueue = new LinkedBlockingQueue<>();

    public static IMqBroker getMqBroker() {
        return mqBroker;
    }
    public static ConsumerRegistrySupport getConsumerRegistrySupport() {
        return consumerRegistrySupport;
    }
    //提供者添加消息
    public static void addMessages(MqMessage mqMessage, boolean async){
        if (async) {
            //异步发送使用多线程并发
            newMessageQueue.add(mqMessage);
        } else {
            //同步发送
            mqBroker.addMessages(Arrays.asList(mqMessage));
        }
    }
    //回调消息
    public static void callbackMessage(MqMessage mqMessage){
        callbackMessageQueue.add(mqMessage);
    }
    //批量添加回调消息
    public static void callbackMessageList(List<MqMessage> msgList) {
        callbackMessageQueue.addAll(msgList);
    }

    //启动Broker服务
    public void startBrokerService() {
        // 初始化 XQRpcInvokerFactory
        xqRpcInvokerFactory = new XQRpcInvokerFactory(XQRegistryServiceRegistry.class, new HashMap<String, String>(){{
            put(XQRegistryServiceRegistry.REGISTRY_ADDRESS, adminAddress);
            put(XQRegistryServiceRegistry.ACCESS_TOKEN, accessToken);
        }});
        try {
            xqRpcInvokerFactory.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        // 初始化注册帮助类
        XQRegistryServiceRegistry commonServiceRegistry = (XQRegistryServiceRegistry) xqRpcInvokerFactory.getServiceRegistry();
        consumerRegistrySupport = new ConsumerRegistrySupport(commonServiceRegistry);

        // 实例化IMqBroker, 动态代理获取对应的消息服务端
        mqBroker = (IMqBroker) new XQRpcReferenceBean(
                NetEnum.NETTY,      //使用Netty做网络通信
                Serializer.SerializeEnum.HESSIAN.getSerializer(),   //使用Hessian序列化数据
                CallType.SYNC,      //使用回调方式
                LoadBalance.ROUND,  //轮训机制
                IMqBroker.class,
                null,
                10000,
                null,
                null,
                null, //new MsgCallBack()
                xqRpcInvokerFactory).getObject();

        //异步多线程添加消息，发送给Broker. 3个线程
        for (int i = 0; i < 3; i++) {
            clientFactoryThreadPool.execute(new Runnable() {
                @Override
                public void run() {

                    while (!MqClientFactory.clientFactoryPoolStoped) {
                        try {
                            MqMessage message = newMessageQueue.take();
                            if (message != null) {
                                //加载
                                List<MqMessage> messageList = new ArrayList<>();
                                messageList.add(message);
                                List<MqMessage> otherMessageList = new ArrayList<>();
                                //一次性取100条数据
                                int drainToNum = newMessageQueue.drainTo(otherMessageList, 49);
                                if (drainToNum > 0) {
                                    messageList.addAll(otherMessageList);
                                }
                                //添加消息到Broker队列中
                                mqBroker.addMessages(messageList);
                                callbackMessageList(messageList);
                            }
                        } catch (Exception e) {
                            if (!MqClientFactory.clientFactoryPoolStoped) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    }

                    //剩余的所有消息
                    List<MqMessage> otherMessageList = new ArrayList<>();
                    int drainToNum = newMessageQueue.drainTo(otherMessageList);
                    if (drainToNum> 0) {
                        mqBroker.addMessages(otherMessageList);
                    }
                    callbackMessageList(otherMessageList);
                }
            });
        }
        //异步多线程添加回调消息，3个线程
        for (int i = 0; i < 3; i++) {
            clientFactoryThreadPool.execute(new Runnable() {
                @Override
                public void run() {

                    while (!MqClientFactory.clientFactoryPoolStoped) {
                        try {
                            MqMessage message = callbackMessageQueue.take();
                            if (message != null) {
                                //加载消息
                                List<MqMessage> messageList = new ArrayList<>();
                                messageList.add(message);

                                List<MqMessage> otherMessageList = new ArrayList<>();
                                int drainToNum = callbackMessageQueue.drainTo(otherMessageList, 49);
                                if (drainToNum > 0) {
                                    messageList.addAll(otherMessageList);
                                }
//                                log.info("执行回调消息数量:" + messageList.size());
                                //回调
                                mqBroker.callbackMessages(messageList);
                            }
                        } catch (Exception e) {
                            if (!MqClientFactory.clientFactoryPoolStoped) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    }

                    //处理剩余的部分
                    List<MqMessage> otherMessageList = new ArrayList<>();
                    int drainToNum = callbackMessageQueue.drainTo(otherMessageList);
                    if (drainToNum> 0) {
                        mqBroker.callbackMessages(otherMessageList);
//                        log.info("执行回调消息数量:" + backMsgCount.addAndGet(otherMessageList.size()));
                    }

                }
            });
        }


    }
    public void destoryBrokerService() throws Exception {
        //停止服务
        if (xqRpcInvokerFactory != null) {
            xqRpcInvokerFactory.stop();
        }
    }


    /**
     * 检查消费者
     */
    private void validConsumer(){
        if (consumerList==null || consumerList.size()==0) {
            log.warn("MqConsumer 消费者不存在.");
            return;
        }

        //构建消费者线程
        for (IMqConsumer consumer : consumerList) {

            //检查消费者注释
            MqConsumer annotation = consumer.getClass().getAnnotation(MqConsumer.class);
            if (annotation == null) {
                throw new RuntimeException(consumer.getClass() +"消费者类, 没有加消息注解.");
            }

            //检查组
            if (annotation.group()==null || annotation.group().trim().length()==0) {
                //如果组为空使用广播消息, 将被uuid替代
                try {
                    //获取成员属性注解数据
                    InvocationHandler invocationHandler = Proxy.getInvocationHandler(annotation);
                    Field mValField = invocationHandler.getClass().getDeclaredField("memberValues");
                    mValField.setAccessible(true);
                    Map memberValues = (Map) mValField.get(invocationHandler);

                    //生成UUID, 放入消息组里
                    String randomGroup = UUID.randomUUID().toString().replaceAll("-", "");
                    memberValues.put("group", randomGroup);
                } catch (Exception e) {
                    throw new RuntimeException(consumer.getClass() +", 组生成失败.");
                }

            }
            if (annotation.group()==null || annotation.group().trim().length()==0) {
                throw new RuntimeException(consumer.getClass() +", 组不能为空.");
            }

            // 检查Topic
            if (annotation.topic()==null || annotation.topic().trim().length()==0) {
                throw new RuntimeException(consumer.getClass() +", 主题不能为空.");
            }

            // 添加到线程集合
            consumerThreadList.add(new ConsumerThread(consumer));
        }
    }
    //启动消费者注册
    private void startConsumer() {
        if (consumerThreadList ==null || consumerThreadList.size()==0) {
            return;
        }

        // 注册消费者
        getConsumerRegistrySupport().registerConsumer(consumerThreadList);

        // 执行消费者线程
        for (ConsumerThread item: consumerThreadList) {
            clientFactoryThreadPool.execute(item);
            log.info("消费者初始化成功！ topic:{}, group:{}", item.getMqConsumer().topic(), item.getMqConsumer().group());
        }

    }

    /**
     * 销毁消费者
     */
    private void destoryConsumer(){
        if (consumerThreadList ==null || consumerThreadList.size()==0) {
            return;
        }
        //移除消费者
        getConsumerRegistrySupport().removeConsumer(consumerThreadList);

    }

}
